package com.atguigu.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.utils.KafkaUtils;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.text.SimpleDateFormat;

/**
 * 从 Kafka 中读取 ods 层 用户行为日志数据
 * <p>
 * 对日志数据进行分流操作
 * <p>
 * 启动日志、曝光日志、页面日志
 * 启动日志放到 启动日志侧输出流中
 * 曝光日志放到 曝光日志侧输出流中
 * 页面日志放到主流中
 * 将不同流的数据写回到 kafka 的 dwd 主题中
 *
 * @author lvbingbing
 * @date 2022-03-12 21:00
 */
public class BaseLogApp {
    public static void main(String[] args) throws Exception {
        // 1．基本环境准备
        // 1.1 流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 1.2 并行度设置
        env.setParallelism(2);
        // 2．检查点相关设置
        // 2.1 开启检查点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        // 2.2 设置检查点超时时间
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        checkpointConfig.setCheckpointTimeout(60000L);
        // 2.3 作业取消时保留检查点
        checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 2.4 设置重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 3000L));
        // 2.5 设置状态后端
        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/ck/gmall"));
        // 2.6 指定操作 HDFS 的用户
        System.setProperty("HADOOP_USER_NAME", "atguigu");
        // 3.从 Kafka 中读取数据
        // Source 的并行度设置为 Kafka 对应 Topic 的分区数
        String topic = "ods_base_log";
        String groupId = "base_log_app_group";
        FlinkKafkaConsumer<String> kafkaConsumer = KafkaUtils.getKafkaSource(topic, groupId);
        DataStreamSource<String> kafkaDs = env.addSource(kafkaConsumer)
                .setParallelism(4);
        // 4．对读取的数据进行结构的转换
        SingleOutputStreamOperator<JSONObject> jsonStreamDs = kafkaDs.map(JSON::parseObject);
        // 5．新老访客状态进行修复
        // 按照设备id进行分组
        SingleOutputStreamOperator<JSONObject> fixedJsonStreamDs = jsonStreamDs.keyBy(jsonObject -> jsonObject.getJSONObject("common").getString("mid"))
                // 使用状态修复新老访客
                .map(new RichMapFunction<JSONObject, JSONObject>() {

                    private static final long serialVersionUID = -5463218433296890050L;

                    /**
                     * 键控状态，保存上一次访问的日期
                     */
                    private transient ValueState<String> visitedState;

                    /**
                     * 日期格式化
                     */
                    private SimpleDateFormat sdf;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        visitedState = getRuntimeContext().getState(new ValueStateDescriptor<String>("visitedState", String.class));
                        sdf = new SimpleDateFormat("yyyyMMdd");
                    }

                    @Override
                    public JSONObject map(JSONObject jsonObject) throws Exception {
                        JSONObject commonObj = jsonObject.getJSONObject("common");
                        String isNew = commonObj.getString("is_new");
                        String currentVisitDate = sdf.format(jsonObject.getLong("ts"));
                        // 获取新老访客状态，只有新访客，才可能需要修复状态
                        if ("1".equals(isNew)) {
                            String visitedDateState = visitedState.value();
                            // 判断状态中的上次访问日期是否为空，如果不为空，可能需要修复状态
                            if (StringUtils.isNotEmpty(visitedDateState)) {
                                // 判断是否在同一天访问，同一天的所有访问都算做新访客
                                if (!currentVisitDate.equals(visitedDateState)) {
                                    commonObj.put("is_new", 0);
                                }
                            } else {
                                // 如果上一次访问日期的状态为空，说明当前是第一次访问
                                visitedState.update(currentVisitDate);
                            }
                        }
                        return jsonObject;
                    }
                });

        // 6．按照日志类型对日志进行分流
        // 启动日志--启动侧输出流   曝光日志--曝光侧输出流    页面--主流
        // 6.1 声明侧输出流标签
        OutputTag<String> startTag = new OutputTag<String>("startTag") {
            private static final long serialVersionUID = 161027306563221106L;
        };
        OutputTag<String> displayTag = new OutputTag<String>("displayTag") {
            private static final long serialVersionUID = 465607865254582214L;
        };
        // 6.2 使用 flink 的侧输出流完成分流
        SingleOutputStreamOperator<String> pageDs = fixedJsonStreamDs.process(new ProcessFunction<JSONObject, String>() {
            private static final long serialVersionUID = -22679605579602881L;

            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, String>.Context ctx, Collector<String> out) throws Exception {
                // 6.2.1 判断是否为启动日志，启动日志放到启动侧输出流
                String jsonObjStr = jsonObject.toJSONString();
                JSONObject startJsonObj = jsonObject.getJSONObject("start");
                if (MapUtils.isNotEmpty(startJsonObj)) {
                    ctx.output(startTag, jsonObjStr);
                } else {
                    // 6.2.2 不是启动日志，那么其他日志都属于页面日志，放到主流
                    out.collect(jsonObjStr);
                    // 判断是否包含曝光日志
                    JSONArray displaysJsonObjArr = jsonObject.getJSONArray("displays");
                    if (CollectionUtils.isNotEmpty(displaysJsonObjArr)) {
                        JSONObject pageJsonStr = jsonObject.getJSONObject("page");
                        String pageId = pageJsonStr.getString("page_id");
                        Long ts = jsonObject.getLong("ts");
                        // 遍历数组，获取每一条曝光日志
                        int size = displaysJsonObjArr.size();
                        for (int i = 0; i < size; i++) {
                            JSONObject displaysJsonObj = displaysJsonObjArr.getJSONObject(i);
                            displaysJsonObj.put("ts", ts);
                            displaysJsonObj.put("page_id", pageId);
                            // 将曝光数据放到曝光侧输出流中
                            ctx.output(displayTag, displaysJsonObj.toJSONString());
                        }
                    }
                }
            }
        });
        // 6.3 获取不同流数据，输出测试
        DataStream<String> startDs = pageDs.getSideOutput(startTag);
        startDs.print("startDs");
        DataStream<String> displayDs = pageDs.getSideOutput(displayTag);
        displayDs.print("displayDs");
        pageDs.print("pageDs");
        // 7．将不同流的数据写 kafka 的 dwd 不同的主题中
        pageDs.addSink(KafkaUtils.getKafkaSinkAtLeastOnce("dwd_page_log"));
        startDs.addSink(KafkaUtils.getKafkaSinkAtLeastOnce("dwd_start_log"));
        displayDs.addSink(KafkaUtils.getKafkaSinkAtLeastOnce("dwd_display_log"));
        // 8、触发程序执行
        env.execute();
    }
}